package streamAPI.sink;

import com.alibaba.druid.util.JdbcUtils;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import util.model.OrderBean;
import util.FlinkConstant;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

/**
 * @author yue.cao
 * @since 10-21-2020
 */
public class DemoWithMySQLSink {


	public static void main(String[] args) throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		DataStreamSource<String> source = env.socketTextStream("caoyuesh1", 4444);

		DataStream<OrderBean> sumRes = source.map(new MapFunction<String, OrderBean>() {
			@Override
			public OrderBean map(String line) throws Exception {
				String[] split = line.split(",");
				return OrderBean.of(split[0], split[1], Double.parseDouble(split[2]));
			}
		}).keyBy("province", "city").sum("money");

		sumRes.addSink(new RichSinkFunction<OrderBean>() {

			private transient Connection connection = null;

			@Override
			public void open(Configuration parameters) throws Exception {
				super.open(parameters);
				connection = DriverManager.getConnection(FlinkConstant._MySQL.URL, FlinkConstant._MySQL.USER_NAME, FlinkConstant._MySQL.PASSWORD);
			}

			@Override
			public void invoke(OrderBean bean, Context context) throws Exception {
				String sql = "INSERT INTO cy_test3(t1,t2,t3) values(?,?,?) ON DUPLICATE KEY UPDATE t3=?";
				PreparedStatement statement = null;
				try {
					statement = connection.prepareStatement(sql);
					statement.setString(1, bean.city);
					statement.setString(2, bean.province);
					statement.setString(3, bean.money.toString());
					statement.setString(4, bean.money.toString());
					statement.executeUpdate();
				} finally {
					JdbcUtils.close(statement);
				}
			}

			@Override
			public void close() throws Exception {
				super.close();
				JdbcUtils.close(connection);
			}
		});
		env.execute();
	}
}
